package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.bean.KeywordStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.IkAnalyzer;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/19 9:10
 */
public class DwsSearchKeyWordApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsSearchKeyWordApp().init(4004, 1, "DwsSearchKeyWordApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
   
        // 1. 建立动态与 topic: dwd_page
        tEnv.executeSql("create table page(" +
                            "   page map<string, string>, " +
                            "   ts bigint, " +
                            "   et as to_timestamp_ltz(ts, 3), " +
                            "   watermark for et as et - interval '3' second" +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            "   'properties.group.id' = 'DwsSearchKeyWordApp', " +
                            "   'topic' = '" + Constant.TOPIC_DWD_PAGE + "', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'json'" +
                            ")");
        // 2. 过滤需要的数据: 找到搜索日志
        /*
          "page": {
            "page_id": "good_list",
            "item": "iphone11",
            "during_time": 11758,
            "item_type": "keyword",
            "last_page_id": "home"
         */
        Table t1 = tEnv.sqlQuery("select" +
                                        " page['item'] kw, " +
                                        " et " +
                                        "from page " +
                                        "where page['page_id'] = 'good_list' " +
                                        "and page['item_type'] = 'keyword' " +
                                        "and page['item'] is not null ");
        tEnv.createTemporaryView("t1", t1);
        // 3. 把关键词分词  自定义函数: TableFunction
        // 3.1 注册函数
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        // 3.2 使用函数
        Table t2 = tEnv.sqlQuery("select" +
                                        " word, " +
                                        " et " +
                                        "from t1 " +
                                        "join lateral table(ik_analyzer(kw)) on true");
        tEnv.createTemporaryView("t2", t2);
    
        // 4. 开窗聚合
        /*
            stt DateTime,
            edt DateTime,
            keyword String ,
            source String ,
            ct UInt64 ,
            ts UInt64
         */
        Table result = tEnv.sqlQuery("select" +
                                        " CONVERT_TZ(date_format(tumble_start(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss'), 'UTC', 'Asia/Shanghai') stt, " +
                                        " CONVERT_TZ(date_format(tumble_end(et, interval '5' second), 'yyyy-MM-dd HH:mm:ss'), 'UTC', 'Asia/Shanghai') edt, " +
                                        " word keyword, " +
                                        " 'search' source, " +
                                        " count(*) ct, " +
                                        " unix_timestamp() *1000 ts " +
                                        "from t2 " +
                                        "group by word, tumble(et, interval '5' second)");
    
        // 5. 数据写入到ClickHouse中
        tEnv
            .toRetractStream(result, KeywordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2022", "keyword_stats_2022", KeywordStats.class));
            
    }
}
